package mr

import (
	"encoding/json"
	"fmt"
	"hash/fnv"
	"io/ioutil"
	"log"
	"net/rpc"
	"os"
	"sort"
	"strconv"
	"time"
)

// Map functions return a slice of KeyValue.
type KeyValue struct {
	Key   string
	Value string
}

// for sorting by key.
type ByKey []KeyValue

// for sorting by key.
func (a ByKey) Len() int           { return len(a) }
func (a ByKey) Swap(i, j int)      { a[i], a[j] = a[j], a[i] }
func (a ByKey) Less(i, j int) bool { return a[i].Key < a[j].Key }

// use ihash(key) % NReduce to choose the reduce
// task number for each KeyValue emitted by Map.
func ihash(key string) int {
	h := fnv.New32a()
	h.Write([]byte(key))
	return int(h.Sum32() & 0x7fffffff)
}

// 读文件
func read_file(filename string) string {
	file, err := os.Open(filename)
	content, err := ioutil.ReadAll(file)
	if err != nil {
		log.Fatalf("cannot read %v", filename)
	}
	return string(content)
}

// main/mrworker.go calls this function.
func Worker(mapf func(string, string) []KeyValue,
	reducef func(string, []string) string) {

	// Your worker implementation here.
	for {
		ask_req := RPCRequestBody{}
		ask_resp := RPCResponseBody{}
		ok := call("Coordinator.TaskAsk", &ask_req, &ask_resp)
		if ok {
			var origin_filename string
			json.Unmarshal(ask_resp.Trans_data, &origin_filename)
			switch ask_resp.Task_type {
			case "map":
				{
					content := read_file(origin_filename)
					kva := mapf(origin_filename, content)
					sort.Sort(ByKey(kva))

					nfiles := []*os.File{}
					for i := 0; i < ask_resp.NReduce; i++ {
						oname := "seq" + strconv.Itoa(ask_resp.Seq) + "-mr-" + strconv.Itoa(ask_resp.Task_id) + "-" + strconv.Itoa(i)
						ofile, _ := os.Create(oname)
						nfiles = append(nfiles, ofile)
					}

					for i := 0; i < len(kva); i++ {
						idx := ihash(kva[i].Key) % ask_resp.NReduce
						enc := json.NewEncoder(nfiles[idx])
						enc.Encode(&kva[i])
					}

					for i := 0; i < len(nfiles); i++ {
						nfiles[i].Sync()
						nfiles[i].Sync()
						nfiles[i].Sync()
						nfiles[i].Close()
					}

					done_req := RPCRequestBody{}
					done_req.Task_id = ask_resp.Task_id
					done_req.Task_type = "map"
					done_req.Trans_data = ask_resp.Trans_data
					done_req.Seq = ask_resp.Seq

					done_resp := RPCResponseBody{}
					_ = call("Coordinator.TaskDone", &done_req, &done_resp)

					if done_resp.Commit {
						for i := 0; i < ask_resp.NReduce; i++ {
							src_name := "seq" + strconv.Itoa(ask_resp.Seq) + "-mr-" + strconv.Itoa(ask_resp.Task_id) + "-" + strconv.Itoa(i)
							dst_name := "mr-" + strconv.Itoa(ask_resp.Task_id) + "-" + strconv.Itoa(i)
							os.Rename(src_name, dst_name)
						}
					} else {
						println("done_resp.Commit failed")
						for i := 0; i < ask_resp.NReduce; i++ {
							src_name := "seq" + strconv.Itoa(ask_resp.Seq) + "-mr-" + strconv.Itoa(ask_resp.Task_id) + "-" + strconv.Itoa(i)
							os.Remove(src_name)
						}
					}

				}
			case "reduce":
				{
					// println("reduce Task_id: " + strconv.Itoa(ask_resp.Task_id) + " Map_count:" + strconv.Itoa(ask_resp.Map_count))
					kvs := []KeyValue{}
					for i := 0; i < ask_resp.Map_count; i++ {
						filename := "mr-" + strconv.Itoa(i) + "-" + strconv.Itoa(ask_resp.Task_id)
						file, _ := os.Open(filename)
						dec := json.NewDecoder(file)
						for {
							var kv KeyValue
							if err := dec.Decode(&kv); err != nil {
								break
							}
							kvs = append(kvs, kv)
						}
						file.Close()
					}

					sort.Sort(ByKey(kvs))

					oname := "seq" + strconv.Itoa(ask_resp.Seq) + "-mr-out-" + strconv.Itoa(ask_resp.Task_id)
					ofile, _ := os.Create(oname)

					i := 0
					for i < len(kvs) {
						j := i + 1
						for j < len(kvs) && kvs[j].Key == kvs[i].Key {
							j++
						}
						values := []string{}
						for k := i; k < j; k++ {
							values = append(values, kvs[k].Value)
						}
						output := reducef(kvs[i].Key, values)

						// this is the correct format for each line of Reduce output.
						fmt.Fprintf(ofile, "%v %v\n", kvs[i].Key, output)

						i = j
					}

					ofile.Sync()
					ofile.Sync()
					ofile.Sync()
					ofile.Close()

					done_req := RPCRequestBody{}
					done_req.Task_id = ask_resp.Task_id
					done_req.Task_type = "reduce"
					done_req.Seq = ask_resp.Seq

					done_resp := RPCResponseBody{}
					_ = call("Coordinator.TaskDone", &done_req, &done_resp)
					if done_resp.Commit {
						src_name := "seq" + strconv.Itoa(ask_resp.Seq) + "-mr-out-" + strconv.Itoa(ask_resp.Task_id)
						dst_name := "mr-out-" + strconv.Itoa(ask_resp.Task_id)
						os.Rename(src_name, dst_name)
					} else {
						println("done_resp.Commit failed")
						src_name := "seq" + strconv.Itoa(ask_resp.Seq) + "-mr-out-" + strconv.Itoa(ask_resp.Task_id)
						os.Remove(src_name)

					}
				}
			case "wait":
				time.Sleep(time.Second * 1)
			case "done":
				os.Exit(0)
			default:
				os.Exit(-1)
			}
		} else {
			fmt.Printf("call failed!\n")
			time.Sleep(10 * time.Second)
		}
	}

	// uncomment to send the Example RPC to the coordinator.
	// CallExample()

}

// example function to show how to make an RPC call to the coordinator.
//
// the RPC argument and reply types are defined in rpc.go.
func CallExample() {

	// declare an argument structure.
	args := ExampleArgs{}

	// fill in the argument(s).
	args.X = 99

	// declare a reply structure.
	reply := ExampleReply{}

	// send the RPC request, wait for the reply.
	// the "Coordinator.Example" tells the
	// receiving server that we'd like to call
	// the Example() method of struct Coordinator.
	ok := call("Coordinator.Example", &args, &reply)
	if ok {
		// reply.Y should be 100.
		fmt.Printf("reply.Y %v\n", reply.Y)
	} else {
		fmt.Printf("call failed!\n")
	}
}

// send an RPC request to the coordinator, wait for the response.
// usually returns true.
// returns false if something goes wrong.
func call(rpcname string, args interface{}, reply interface{}) bool {
	// c, err := rpc.DialHTTP("tcp", "127.0.0.1"+":1234")
	sockname := coordinatorSock()
	c, err := rpc.DialHTTP("unix", sockname)
	if err != nil {
		log.Fatal("dialing:", err)
	}
	defer c.Close()

	err = c.Call(rpcname, args, reply)
	if err == nil {
		return true
	}

	fmt.Println(err)
	return false
}
